package se.kth.ict.id2203.epfd;

import java.util.HashSet;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import se.kth.ict.id2203.pp2p.PerfectPointToPointLink;
import se.kth.ict.id2203.pp2p.Pp2pSend;
import se.kth.ict.id2203.flp2p.FairLossPointToPointLink;
import se.sics.kompics.ComponentDefinition;
import se.sics.kompics.Handler;
import se.sics.kompics.Negative;
import se.sics.kompics.Positive;
import se.sics.kompics.address.Address;
import se.sics.kompics.launch.Topology;
import se.sics.kompics.timer.ScheduleTimeout;
import se.sics.kompics.timer.Timer;


public class EPFD extends ComponentDefinition {
	
	Negative<EPerfectFailureDetector> epfdPort = provides(EPerfectFailureDetector.class);
	
	Positive<PerfectPointToPointLink> pp2p = requires(PerfectPointToPointLink.class);	
	Positive<FairLossPointToPointLink> flp2p = requires(FairLossPointToPointLink.class);
	
	Positive<Timer> timer = requires(Timer.class);

	
	private Address self;
	private Topology topology;
	private Set<Address> alive;
	private Set<Address> suspected;
	private Set<Address> neighborSet;
	private long timeDelay;
	private long delta;
	private long period;

	private static final Logger logger =
			LoggerFactory.getLogger(EPFD.class);
	
	public EPFD() {
		logger.info("EPFD created.");
		subscribe(handleInit, control);
		subscribe(handleCheckTimeout, timer);
		subscribe(handleHeartbeatTimeout, timer);
		subscribe(handleHeartbeatMessage, pp2p);
		subscribe(handleHeartbeatMessage2, flp2p);
	}

	Handler<EPFDInit> handleInit = new Handler<EPFDInit>() {
		public void handle(EPFDInit event) {
			
			logger.info("EPFD started.");
			topology = event.getTopology();
			self = topology.getSelfAddress();
			alive = topology.getNeighbors(self);
			suspected = new HashSet<Address>();
			neighborSet = topology.getNeighbors(self);
			timeDelay = event.getTimeDelay();
			delta = event.getDelta();
			period = timeDelay;
			
			// startTimer for heartbeant
			ScheduleTimeout sthb = new ScheduleTimeout(timeDelay);
			sthb.setTimeoutEvent(new HeartbeatTimeout(sthb));
			trigger(sthb, timer);
			// startTimer for checking
			ScheduleTimeout stck = new ScheduleTimeout(period);
			stck.setTimeoutEvent(new CheckTimeout(stck));
			trigger(stck, timer);			
			
		}
	};
	
	Handler<CheckTimeout> handleCheckTimeout = new Handler<CheckTimeout>() {
		public void handle(CheckTimeout event) {
			// do algorithm and deliver suspect or restore event
			logger.info("Checking neighbors...");
			
			Set<Address> temp = new HashSet<Address>(alive);
			temp.retainAll(suspected);
			if (temp.size() > 0) {
				period = period + delta;
			}
			
			for (Address neighbor : neighborSet) {

				if ((!alive.contains(neighbor)) && (!suspected.contains(neighbor))) {
					suspected.add(neighbor);
					trigger(new Suspect(neighbor, period), epfdPort);
					
				} else if ((alive.contains(neighbor)) && (suspected.contains(neighbor))) {
					suspected.remove(neighbor);
					trigger(new Restore(neighbor, period), epfdPort);
				}
				
			}
			
			alive.clear();
			
			ScheduleTimeout stck = new ScheduleTimeout(period);
			stck.setTimeoutEvent(new CheckTimeout(stck));
			trigger(stck, timer);	
			
		}
	};
	
	Handler<HeartbeatTimeout> handleHeartbeatTimeout = new Handler<HeartbeatTimeout>() {
		public void handle(HeartbeatTimeout event) {
			// broadcast HeartbeatMessage
			for (Address neighbor : neighborSet) {
			
			/*
			 * trigger HeartbeatMessage through pp2p link or flp2p link
			 * choose one below
			 */
							
				trigger(new Pp2pSend(neighbor, new HeartbeatMessage(self)),	pp2p);
			//	trigger(new Flp2pSend(neighbor, new HeartbeatMessage2(self)), flp2p);
				
			}
			
			logger.info("Broadcasting HeartBeat Message...");
			
			ScheduleTimeout sthb = new ScheduleTimeout(timeDelay);
			sthb.setTimeoutEvent(new HeartbeatTimeout(sthb));
			trigger(sthb, timer);
			
		}
	};
	
	Handler<HeartbeatMessage> handleHeartbeatMessage = new Handler<HeartbeatMessage>() {
		public void handle(HeartbeatMessage event) {
			// handle HeartbeatMessage from pp2p link
			Address src = event.getSource();
			if (!alive.contains(src)) {
				alive.add(src);
			}
			logger.info("HeartBeat Message from " + src);
		}
	};
	
	Handler<HeartbeatMessage2> handleHeartbeatMessage2 = new Handler<HeartbeatMessage2>() {
		public void handle(HeartbeatMessage2 event) {
			// handle HeartbeatMessage from flp2p link
			Address src = event.getSource();
			if (!alive.contains(src)) {
				alive.add(src);
			}
			logger.info("HeartBeat Message from " + src);
		}
	};
	
}
